package cn.simon.free.redis.mq.core.consumer;

import cn.simon.free.redis.mq.core.AbstractRedisMessage;
import cn.simon.free.redis.mq.enums.ConsumerType;
import cn.simon.free.redis.mq.model.Message;
import com.google.common.reflect.TypeToken;

import java.lang.reflect.ParameterizedType;

/**
 * 次级抽象类,负责拉取延迟队列里面的数据
 *
 * @author simon
 * @date 2021/7/1 14:20
 * @copyright:
 * @modifier
 * @motto a no talent, people who have to study hard
 */
public abstract class AbstractRedisMessageConsumer<T> extends AbstractRedisMessage {

    protected final Class<T> convertClass = (Class) ((ParameterizedType) TypeToken.of(this.getClass()).getRawType().getGenericSuperclass()).getActualTypeArguments()[0];

    /**
     * 获取redis消息队列信息
     *
     * @param queueName
     * @param consumerType
     * @return
     */
    public abstract Message getMessage(String queueName, ConsumerType consumerType);

    /**
     * 删除消息
     *
     * @param queueName
     * @param message
     */
    public abstract void deleteMessage(String queueName, Message message);

    /**
     * 重新放回队列,当 process() 子类的方法返回为false时,且集群模式{@link ConsumerType.CLUSTER}下,默认会触发该方法（重推数据回消息队列）,
     * 如果数据都是一次性消费,不需要再放回队列,可以按照下列方式处理
     * 方式: process()方法永远返回true
     * PS: 广播模式(BROADCAST),不会触发该方法
     *
     * @param queueName
     * @param message
     * @param consumerType
     */
    public abstract void rePushMessage(String queueName, Message message, ConsumerType consumerType);

    /**
     * 子类实现,用户消费类(必须可重写)
     *
     * @param t
     * @param message
     * @return
     */
    protected abstract boolean process(T t, Message message);

    /**
     * 数据转换对象,转换器默认为jackson(可重写)
     *
     * @param message
     * @return
     */
    protected T parseMessage(Message message) {
        return mapper.convertValue(message.getMessage(), convertClass);
    }

    /**
     * 入口,内嵌方法
     *
     * @param consumerQueue
     * @param message
     * @param consumerType
     * @return
     */
    public boolean processMessage(String consumerQueue, Message message, ConsumerType consumerType) {
        T t = this.parseMessage(message);
        if (this.process(t, message)) {
            return true;
        }
        rePushMessage(consumerQueue, message, consumerType);
        return false;
    }

}
